package com.ionehe.group.common.notify.rocketmq.producer;

import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

import com.beust.jcommander.Strings;
import com.ionehe.group.common.notify.autoconfigure.AsyncVendors;
import com.ionehe.group.common.notify.configure.NotifyConfiguration;
import com.ionehe.group.common.notify.message.exception.AsyncException;
import com.ionehe.group.common.notify.message.producer.Producer;
import com.ionehe.group.common.notify.rocketmq.flags.RocketKeyFlags;
import com.ionehe.group.common.notify.util.Json;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingConnectException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;

/**
 * Copyright (c) 2021 ionehe.com
 * Date: 2021/4/21
 * Time: 1:22 下午
 *
 * @author 2021年 <a href="mailto:a@ionehe.com">秀</a>
 * 【rocketmq生产者】
 */
@Slf4j
public class ProducerImpl<T> implements Producer<T> {

    private final DefaultMQProducer defaultMQProducer;
    private final AtomicBoolean started = new AtomicBoolean(false);
    private final AtomicBoolean closed = new AtomicBoolean(false);

    private Map<String, String> topics;

    public ProducerImpl(NotifyConfiguration conf) {
        this.topics = conf.getTopics();
        Map<String, Object> vendor = (Map<String, Object>) conf.getVendors().get(AsyncVendors.rocketmq);
        this.defaultMQProducer = new DefaultMQProducer((String) RocketKeyFlags.FLAG_PRODUCER_GROUP.get(vendor));
        this.defaultMQProducer.setSendMsgTimeout(((Integer) RocketKeyFlags.FLAG_SEND_MSG_TIMEOUT.get(vendor)).intValue());
        this.defaultMQProducer.setInstanceName((String) RocketKeyFlags.FLAG_INSTANCE_NAME.get(vendor));
        String nameServer = (String) RocketKeyFlags.FLAG_NAMESRVADDR.get(vendor);
        if (!Strings.isStringEmpty(nameServer)) {
            this.defaultMQProducer.setNamesrvAddr(nameServer);
        }
        this.defaultMQProducer.setMaxMessageSize(((Integer) RocketKeyFlags.FLAG_MSG_SIZE.get(vendor)).intValue());
    }


    @Override
    public void start() {
        try {
            if (this.started.compareAndSet(false, true)) {
                this.defaultMQProducer.start();
            }
        } catch (Exception e) {
            throw new AsyncException(e.getMessage(), e);
        }
    }


    @Override
    public void shutdown() {
        if (this.closed.compareAndSet(false, true)) {
            this.defaultMQProducer.shutdown();
        }
    }


    @Override
    public SendResult send(Message message) {
        checkServiceState();

        try {
            return this.defaultMQProducer.send(message);
        } catch (Exception e) {
            log.error(String.format("Send message exception, %s", new Object[]{message}), e);
            checkProducerException(e, message);
            return null;
        }
    }


    @Override
    public void sendOneway(Message message) {
        checkServiceState();
        try {
            this.defaultMQProducer.sendOneway(message);
        } catch (Exception e) {
            log.error(String.format("Send message oneway exception, %s", new Object[]{message}), e);
            checkProducerException(e, message);
        }
    }


    @Override
    public SendResult send(String model, String tags, String keys, T message) {
        checkServiceState();

        byte[] data = Json.of(message);
        Message msg = new Message(this.topics.get(model), tags, keys, data);
        try {
            return this.defaultMQProducer.send(msg);
        } catch (Exception e) {
            log.error(String.format("Send message Exceptions %s", new Object[]{msg}), e);
            checkProducerException(e, msg);
            return null;
        }
    }


    @Override
    public void sendOneway(String model, String tags, String keys, T message) {
        checkServiceState();

        byte[] data = Json.of(message);
        Message msg = new Message(this.topics.get(model), tags, keys, data);
        try {
            this.defaultMQProducer.sendOneway(msg);
        } catch (Exception e) {
            log.error(String.format("Send message oneway exception, %s", new Object[]{message}), e);
            checkProducerException(e, msg);
        }
    }


    @Override
    public boolean isStarted() {
        return this.started.get();
    }


    @Override
    public boolean isClosed() {
        return this.closed.get();
    }


    private void checkServiceState() {
        switch (this.defaultMQProducer.getDefaultMQProducerImpl().getServiceState()) {
            case CREATE_JUST:
                throw new AsyncException(String.format("You do not have start the producer, %s", new Object[]{this.defaultMQProducer
                        .getDefaultMQProducerImpl().getServiceState()}));
            case SHUTDOWN_ALREADY:
                throw new AsyncException(String.format("Your producer has been shut down, %s", new Object[]{this.defaultMQProducer
                        .getDefaultMQProducerImpl().getServiceState()}));
            case START_FAILED:
                throw new AsyncException(String.format("When you start your service throws an exception, %s", new Object[]{this.defaultMQProducer
                        .getDefaultMQProducerImpl().getServiceState()}));
            default:
        }
    }


    private void checkProducerException(Exception e, Message message) {
        if (e instanceof MQClientException) {
            if (e.getCause() != null) {
                if (e.getCause() instanceof RemotingConnectException) {
                    throw new AsyncException(String.format("Connect broker failed, Topic: %s", new Object[]{message.getTopic()}));
                }

                if (e.getCause() instanceof RemotingTimeoutException) {
                    throw new AsyncException(String.format("Send message to broker timeout, %dms, Topic: %s", new Object[]{
                            Integer.valueOf(this.defaultMQProducer.getSendMsgTimeout()), message.getTopic()
                    }));
                }

                if (e.getCause() instanceof MQBrokerException) {
                    MQBrokerException excep = (MQBrokerException) e.getCause();
                    throw new AsyncException(String.format("Receive a broker exception, Topic: %s, %s", new Object[]{message
                            .getTopic(), excep.getErrorMessage()}));
                }

            } else {

                MQClientException excep = (MQClientException) e;
                if (-1 == excep.getResponseCode()) {
                    throw new AsyncException(String.format("Topic does not exist, Topic: %s, %s", new Object[]{message
                            .getTopic(), excep.getErrorMessage()}));
                }
                if (13 == excep.getResponseCode()) {
                    throw new AsyncException(String.format("ons Client check message exception, Topic: %s, %s", new Object[]{message
                            .getTopic(), excep.getErrorMessage()}));
                }
            }
            throw new AsyncException("defaultMQProducer send exception",e);
        }
    }


}
